/*
 * Copyright 2011-2012 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.sabayframework.cluster.impl;

import java.util.concurrent.ThreadFactory;

import org.jgroups.logging.Log;
import org.jgroups.logging.LogFactory;
import org.sabayframework.activity.Activity;
import org.sabayframework.activity.ActivityMonitor;
import org.sabayframework.channel.Channel;
import org.sabayframework.cluster.ClusterId;
import org.sabayframework.cluster.Computer;
import org.sabayframework.cluster.ComputerId;
import org.sabayframework.cluster.DistributedActivity;
import org.sabayframework.cluster.NodeExecutionQueue;
import org.sabayframework.cluster.NodeId;
import org.sabayframework.cluster.NodeMapping;
import org.sabayframework.cluster.ProcessorId;
import org.sabayframework.queue.impl.DefaultThreadFactory;


public class LocalComputer extends BaseNode<ComputerId, ClusterId> implements Computer, NodeExecutionQueue {

	private final LocalProcessor[] processors;
	private final ModuloNodeMapping<ProcessorIdImpl> nodeMapping;
	private static final Log log = LogFactory.getLog(LocalComputer.class);
	
	public LocalComputer(ComputerIdImpl id) {
		this(id, Runtime.getRuntime().availableProcessors());
	}
	public LocalComputer(ComputerIdImpl id, int numberOfProcessors) {
		this(id, numberOfProcessors, new DefaultThreadFactory());
	}

	public LocalComputer(ComputerIdImpl id, int numberOfProcessors, ThreadFactory threadFactory) {
		super(id);
		if(numberOfProcessors <=0 || numberOfProcessors > 1023)
			throw new IllegalArgumentException("Number of processors is out of 1...1023 range:  "+numberOfProcessors);
		processors=new LocalProcessor[numberOfProcessors];
		final ProcessorIdImpl[] ids = new ProcessorIdImpl[numberOfProcessors];
		for(int i=0;i<numberOfProcessors;++i){
			ids[i] = new ProcessorIdImpl(id, i);
			processors[i] = new LocalProcessor(ids[i], threadFactory, 10);
		}
		nodeMapping = new ModuloNodeMapping<ProcessorIdImpl>(ids);
	}
	
	@Override
	public ComputerId getComputerId() {
		return id;
	}
	
	public void start(){
		for(LocalProcessor p: processors){
			try{
				p.start();
			}catch (Exception e) {
				log.error("Start "+p.toString(), e);
			}
		}
		if(log.isDebugEnabled()) log.debug(getId()+" started.");
	}
	
	public void stop(){
		for(LocalProcessor p: processors){
			try{
				p.stop();
			}catch (Exception e) {
				log.error("Stop "+p.toString(), e);
			}
		}
		if(log.isDebugEnabled()) log.debug(getId()+" stopped.");
	}

	public NodeId[] getChildrenIds() {
		final NodeId[] ids=new NodeId[processors.length];
		for(int i=0;i<ids.length;++i) ids[i] = processors[i].getId();
		return ids;
	}

	public int getChildrenCount() {
		return processors.length;
	}

	public NodeMapping getChildrenMapping() {
		return nodeMapping;
	}
	@Override
	public int size() {
		int size = 0;
		for(int i=0;i<processors.length;++i) size+= processors[i].getExecutionQueue().size();
		return size;
	}
	@Override
	public boolean isEmpty() {
		for(int i=0;i<processors.length;++i) 
			if(!processors[i].getExecutionQueue().isEmpty()) return false;
		return true;
	}
	@Override
	public void channelChanged(Channel<?> input) {
		for(int i=0;i<processors.length;++i) 
			processors[i].getExecutionQueue().channelChanged(input);
		
	}
	
	@Override
	public NodeExecutionQueue getExecutionQueue() {
		return this;
	}
	@Override
	public int workerCount() {
		return processors.length;
	}
	@Override
	public <I, O> ActivityMonitor<I, O> submit(Activity<I, O> activity,	Channel<I> input, Channel<O> output) {
		ProcessorId nodeId = nodeMapping.get(activity.hashCode());
		return processors[nodeId.getProcessorNo()].getExecutionQueue().submit(activity, input, output);
	}
	@Override
	public <I, O> ActivityMonitor<I, O> execute(Activity<I, O> activity, Channel<I> input, Channel<O> output) {
		ProcessorId nodeId = nodeMapping.get(activity.hashCode());
		return processors[nodeId.getProcessorNo()].getExecutionQueue().execute(activity, input, output);
	}
	@Override
	public <I, O> ActivityMonitor<I, O> submit(DistributedActivity<I, O> da, Channel<I> input, Channel<O> output) {
		NodeId nodeId = da.getDistributionStrategy().select(this, da, null);
		if(!(nodeId instanceof ProcessorId)) throw new IllegalArgumentException("Invalid children id: "+nodeId);
		ProcessorId procId = (ProcessorId) nodeId; 
		return processors[procId.getProcessorNo()].getExecutionQueue().execute(da.getActivity(), input, output);
	}


}
